package org.bigdata.leo.hbase.starter.sdk;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.bigdata.leo.hbase.starter.model.HBaseConnThreadPoolInit;
import org.bigdata.leo.hbase.starter.model.HBaseTableDesc;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
import org.springframework.util.StopWatch;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * @author leo.jie (weixiao.me@aliyun.com)
 * @version 1.0
 * @organization bigdata
 * @website https://www.jielongping.com
 * @date 2020/3/30 9:54 下午
 * @since 1.0
 */
public class HBaseTemplate implements HBaseOperations {
    private static final Logger LOGGER = LoggerFactory.getLogger(HBaseTemplate.class);

    private Configuration configuration;
    private HBaseConnThreadPoolInit hBaseConnThreadPoolInit;

    private volatile Connection connection;

    public HBaseTemplate(Configuration configuration, HBaseConnThreadPoolInit hBaseConnThreadPoolInit) {
        this.setConfiguration(configuration);
        this.sethBaseConnThreadPoolInit(hBaseConnThreadPoolInit);
        Assert.notNull(configuration, " a valid configuration is provided.");
    }

    @Override
    public boolean createTable(HBaseTableDesc tableDesc) {
        return this.execute(admin -> {
            TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableDesc.getTableName()))
                    .setColumnFamilies(tableDesc.getFamilies().stream().map(family ->
                            ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family))
                                    .build()).collect(Collectors.toList()))
                    .build();
            admin.createTable(tableDescriptor);
            return true;
        });
    }

    @Override
    public boolean createNamespace(String namespace) {
        return this.execute(admin -> {
            admin.createNamespace(NamespaceDescriptor.create(namespace).build());
            return true;
        });
    }

    @Override
    public boolean deleteTable(String tableName) {
        return this.execute(admin -> {
            admin.deleteTable(TableName.valueOf(tableName));
            return true;
        });
    }

    @Override
    public boolean enableTable(String tableName) {
        return this.execute(admin -> {
            admin.enableTable(TableName.valueOf(tableName));
            return true;
        });
    }

    @Override
    public boolean disableTable(String tableName) {
        return this.execute(admin -> {
            admin.disableTable(TableName.valueOf(tableName));
            return true;
        });
    }

    @Override
    public List<String> getTableNames() {
        return this.execute(admin -> {
            final TableName[] tableNames = admin.listTableNames();
            return Arrays.stream(tableNames).map(TableName::getNameAsString).collect(Collectors.toList());
        });
    }

    @Override
    public List<String> getSchemas() {
        return this.execute(admin -> {
            final NamespaceDescriptor[] listNamespaceDescriptors = admin.listNamespaceDescriptors();
            return Arrays.stream(listNamespaceDescriptors)
                    .map(NamespaceDescriptor::getName)
                    .collect(Collectors.toList());
        });
    }


    @Override
    public <T> T execute(AdminCallback<T> adminAction) {
        Assert.notNull(adminAction, "Admin Action object must not be null");
        StopWatch sw = new StopWatch();
        sw.start();
        Admin admin = null;
        try {
            admin = this.getConnection().getAdmin();
            return adminAction.doInAdmin(admin);
        } catch (Throwable throwable) {
            throw new HBaseSystemException(throwable);
        } finally {
            if (null != admin) {
                try {
                    admin.close();
                    sw.stop();
                } catch (IOException e) {
                    LOGGER.error("HBase admin resource release failed.");
                }
            }
        }
    }

    @Override
    public <T> T execute(String tableName, TableCallback<T> action) {
        Assert.notNull(action, "The table callback must not be null.");
        Assert.notNull(tableName, "no table be provided.");

        StopWatch sw = new StopWatch();
        sw.start();
        Table table = null;
        try {
            table = this.getConnection().getTable(TableName.valueOf(tableName));
            return action.doInTable(table);
        } catch (Throwable throwable) {
            throw new HBaseSystemException(throwable);
        } finally {
            if (null != table) {
                try {
                    table.close();
                    sw.stop();
                } catch (IOException e) {
                    LOGGER.error("HBase table resource release failed.");
                }
            }
        }
    }

    @Override
    public <T> List<T> find(String tableName, String family, RowMapper<T> action) {
        Scan scan = new Scan();
        scan.setCaching(5000);
        scan.addFamily(Bytes.toBytes(family));
        return this.find(tableName, scan, action);
    }

    @Override
    public <T> List<T> find(String tableName, String family, String qualifier, RowMapper<T> action) {
        Scan scan = new Scan();
        scan.setCaching(5000);
        scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
        return this.find(tableName, scan, action);
    }

    @Override
    public <T> List<T> find(String tableName, Scan scan, RowMapper<T> action) {
        return this.execute(tableName, table -> {
            int caching = scan.getCaching();
            // if not set caching,change its value to 5000
            if (caching == 1) {
                scan.setCaching(5000);
            }
            try (ResultScanner scanner = table.getScanner(scan)) {
                List<T> rs = new ArrayList<>();
                int rowNum = 0;
                for (Result result : scanner) {
                    rs.add(action.mapRow(result, rowNum++));
                }
                return rs;
            }
        });
    }

    @Override
    public <T> T get(String tableName, String rowName, RowMapper<T> mapper) {
        return this.get(tableName, rowName, null, null, mapper);
    }

    @Override
    public <T> T get(String tableName, String rowName, String familyName, RowMapper<T> mapper) {
        return this.get(tableName, rowName, familyName, null, mapper);
    }

    @Override
    public <T> T get(String tableName, String rowName, String familyName, String qualifier, RowMapper<T> mapper) {
        return this.execute(tableName, table -> {
            Get get = new Get(Bytes.toBytes(rowName));
            if (StringUtils.isNotBlank(familyName)) {
                byte[] family = Bytes.toBytes(familyName);
                if (StringUtils.isNotBlank(qualifier)) {
                    get.addColumn(family, Bytes.toBytes(qualifier));
                } else {
                    get.addFamily(family);
                }
            }
            Result result = table.get(get);
            return mapper.mapRow(result, 0);
        });
    }

    @Override
    public void execute(String tableName, MutatorCallback action) {
        Assert.notNull(action, "MutatorCallback object must not be null");
        Assert.notNull(tableName, "No table provided.");

        StopWatch sw = new StopWatch();
        sw.start();
        BufferedMutator mutator = null;
        try {
            BufferedMutatorParams mutatorParams = new BufferedMutatorParams(TableName.valueOf(tableName));
            mutator = this.getConnection().getBufferedMutator(mutatorParams.writeBufferSize(3 * 1024 * 1024));
            action.doInMutator(mutator);
        } catch (Throwable throwable) {
            throw new HBaseSystemException(throwable);
        } finally {
            if (null != mutator) {
                try {
                    mutator.flush();
                    mutator.close();
                    sw.stop();
                } catch (IOException e) {
                    LOGGER.error("HBase resources release error.");
                }
            }
        }
    }

    @Override
    public void saveOrUpdate(String tableName, Mutation mutation) {
        this.execute(tableName, mutator -> {
            mutator.mutate(mutation);
        });
    }

    @Override
    public void saveOrUpdates(String tableName, List<Mutation> mutations) {
        this.execute(tableName, mutator -> {
            mutator.mutate(mutations);
        });
    }

    public void setConnection(Connection connection) {
        this.connection = connection;
    }

    public Connection getConnection() {
        if (null == this.connection) {
            synchronized (this) {
                if (null == this.connection) {
                    try {
                        ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(hBaseConnThreadPoolInit.getCorePoolSize(),
                                hBaseConnThreadPoolInit.getMaximumPoolSize(), hBaseConnThreadPoolInit.getKeepAliveTime(),
                                TimeUnit.SECONDS, hBaseConnThreadPoolInit.getWorkQueue(), hBaseConnThreadPoolInit.hBasePoolThreadFactory());
                        //ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(20, Integer.MAX_VALUE, 60L,
                        // TimeUnit.SECONDS, new SynchronousQueue<>());
                        // init pool
                        poolExecutor.prestartCoreThread();
                        this.connection = ConnectionFactory.createConnection(configuration, poolExecutor);
                    } catch (IOException e) {
                        LOGGER.error("HBase connection pool created error.");
                    }
                }
            }
        }
        return this.connection;
    }

    public Configuration getConfiguration() {
        return configuration;
    }

    public void setConfiguration(Configuration configuration) {
        this.configuration = configuration;
    }

    public HBaseConnThreadPoolInit gethBaseConnThreadPoolInit() {
        return hBaseConnThreadPoolInit;
    }

    public void sethBaseConnThreadPoolInit(HBaseConnThreadPoolInit hBaseConnThreadPoolInit) {
        this.hBaseConnThreadPoolInit = hBaseConnThreadPoolInit;
    }
}
